package com.nathan.mq.integration.vender.redismq;

import com.nathan.mq.integration.constant.MessageModel;
import com.nathan.mq.integration.service.RedisService;
import com.nathan.mq.integration.vender.redismq.annotation.RedisQueueListener;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RTopic;
import org.springframework.aop.framework.AopProxyUtils;
import org.springframework.context.ApplicationContext;

/**
 * @Description:
 * @Author: zhujunjie
 * @Date: 2021/12/9 17:20
 */
@Slf4j
public class RedisQueueBean<T> {

    private final ApplicationContext applicationContext;
    private final Map<String, RedisMqListener> map;
    private final Map<String, Thread> threadMap = new HashMap<>(64);

    RedisQueueBean(Map<String, RedisMqListener> map, ApplicationContext applicationContext) {
        this.map = map;
        this.applicationContext = applicationContext;
    }


    public void initQueueBean() {
        AtomicInteger count = new AtomicInteger();
        map.forEach((beanName, bean) -> {
            Class<?> clazz = AopProxyUtils.ultimateTargetClass(bean);

            if (RedisQueueListener.class.isAssignableFrom(bean.getClass())) {
                throw new IllegalStateException(clazz + " cannot be both instance of " + RedisQueueListener.class.getName());
            }

            RedisQueueListener annotation = clazz.getAnnotation(RedisQueueListener.class);
            //队列
            if (MessageModel.Blockqueue.getModeCN().equals(annotation.messageModel().getModeCN())) {

                threadMap.putIfAbsent(annotation.name(), new Thread(() -> {
                    final Thread thread = Thread.currentThread();
                    thread.setName("redis-queue-thread-" + count.incrementAndGet());
                    boolean initFlag = false;
                    final RBlockingQueue<String> blockingQueue = applicationContext.getBean(RedisService.class).getBlockingQueue(annotation.name());
                    while (true) {
                        if (!initFlag) {
                            initFlag = true;
                            log.info("注册队列监听 主线程:{} 绑定延迟队列:{} 成功", thread.getId(), annotation.name());
                        }
                        try {
                            final String msg = blockingQueue.take();
                            try {
                                bean.receiveMsg(msg);
                            } catch (Exception e) {
                                log.error("注册队列监听 队列:{} 数据:{} 执行异常:", annotation.name(), msg, e);
                            }
                            if (blockingQueue.size() == 0) {
                                TimeUnit.MILLISECONDS.sleep(500);
                            }
                            //finally {
                            //    MDC.remove(RedisQueueUtil.MESSAGE_ID);
                            //}
                        } catch (InterruptedException e) {
                            log.error("注册队列监听 主线程:{} 获取消息中断异常:", thread.getId(), e);
                            break;
                        }
                    }
                    log.info("注册队列监听 主线程:{} 关闭", thread.getId());
                }));
            } else {
                //广播
                final RTopic topic = applicationContext.getBean(RedisService.class).getTopic(annotation.name(),
                    cn.hutool.core.util.ReflectUtil.newInstance(annotation.codec()));
                topic.addListener(annotation.msgType(), (charSequence, data) -> {
                    try {
                        bean.receiveMsg(data);
                    } catch (Exception e) {
                        log.error("Redis发布订阅 error:", e);
                    }
                });
                log.info("Redis发布订阅 注册Topic:{} {} 成功", annotation.name(), topic.getChannelNames());
            }
        });
        threadMap.forEach((k, v) -> v.start());
    }

    void removeAllListener() {
        threadMap.forEach((k, v) -> v.interrupt());
    }

    void removeSubscribeListener() {
        map.forEach((beanName, bean) -> {
            final RedisQueueListener annotation = bean.getClass().getAnnotation(RedisQueueListener.class);
            if (annotation == null) {
                return;
            }
            if (MessageModel.Subscribe.getModeCN().equals(annotation.messageModel().getModeCN())) {
                final RTopic topic = applicationContext.getBean(RedisService.class).getTopic(annotation.name(),
                    cn.hutool.core.util.ReflectUtil.newInstance(annotation.codec()));
                topic.removeAllListeners();
                log.info("Redis发布订阅 移除Topic:{} {} 成功", annotation.name(), topic);
            }
        });
    }

}
